#include "pch.h"
#include "SyncData.h"
#include <string>
#include <map>
#include "../DataModule/HanaData.h"
#include "SyncDataCache.h"
#include "Util.h"

std::map<std::string, std::shared_ptr<SyncData>> SyncData::sync_datas_map = std::map<std::string, std::shared_ptr<SyncData>>();

SyncData::SyncData(std::shared_ptr<EnclaveProvider> &enclave_provider, Config *config)
    : enclave_provider_(enclave_provider), config_(config), event_(true, false, "sync_data")
{
}


SyncData::~SyncData()
{
}

void SyncData::generatorTask(HanaData &hana, std::vector<std::string> &tables, enclave::SyncTablesDatas &sync_tables_datas, enclave::SyncTablesDatas &sync_info_request)
{
	std::map<std::string, SyncDataCache::TableDatas> tables_cache_datas;
	for (int i = 0; i < sync_tables_datas.tablesyncdatas_size(); i++)
	{
		std::string table = sync_tables_datas.tablesyncdatas(i).table();

		SyncDataCache::TableDatas table_datas;
		table_datas.end_ = sync_tables_datas.tablesyncdatas(i).end();

		for (int j = 0; j < sync_tables_datas.tablesyncdatas(i).syncdatas_size(); j++)
		{
			SyncDataCache::Data data;
			data.start_ = sync_tables_datas.tablesyncdatas(i).syncdatas(j).start();
			data.end_ = sync_tables_datas.tablesyncdatas(i).syncdatas(j).end();
			table_datas.datas_.push_back(data);
		}

		tables_cache_datas[table] = table_datas;
	}

	IData::Fields fields;
	for (auto iter = tables.begin(); iter != tables.end(); iter++)
	{
		std::string sql = "select min(seq), max(seq) from ";
		sql += *iter + ";";

		int num_rows;
		if (hana.select(sql, fields, num_rows, nullptr) == false || num_rows == 0)
		{
			LOG(ERROR) << sql << ", num_rows = " << num_rows;
			continue;
		}

		int min = atoi(std::string(fields.buf(), fields.field_len(0)).c_str());
		int max = atoi(std::string(fields.buf() + fields.field_len(0), fields.field_len(1)).c_str());

        auto table_sync_data = sync_info_request.add_tablesyncdatas();
        table_sync_data->set_table(*iter);
        table_sync_data->set_end(max);

		if (tables_cache_datas.find(*iter) != tables_cache_datas.end())
		{
			min = tables_cache_datas[*iter].end_;

			for (auto data_iter = tables_cache_datas[*iter].datas_.begin(); data_iter < tables_cache_datas[*iter].datas_.end(); data_iter++)
			{
				Task task;
				task.table_ = *iter;
				task.start_ = data_iter->start_;
				task.end_ = data_iter->end_;
				tasks_.push_back(task);

                auto sync_data = table_sync_data->add_syncdatas();
                sync_data->set_start(task.start_);
                sync_data->set_end(task.end_);
			}
		}

		while (min < max)
		{
			Task task;
			task.table_ = *iter;
			task.start_ = min;
			
			int temp = min + config_->read_count_;
			min = temp >= max ? max + 1 : temp;

			task.end_ = min - 1;

			tasks_.push_back(task);

            auto sync_data = table_sync_data->add_syncdatas();
            sync_data->set_start(task.start_);
            sync_data->set_end(task.end_);
		}
	}
}

void SyncData::syncData(const std::string &sync_id, Vault &vault, std::shared_ptr<EnclaveProvider> &enclave_provider)
{
    LOG(INFO) << "Create sync thread";

    for (int i = 0; i < config_->thread_number_; i++)
    {
        auto sync_data_thread = std::shared_ptr<SyncDataThread>(new SyncDataThread(sync_id,this, &vault, enclave_provider));

        if (sync_data_thread->run())
        {
            sync_data_threads_.push_back(sync_data_thread);
        }
    }

    LOG(INFO) << "Sync number of threads is " << sync_data_threads_.size();

    HANDLE *handle = new HANDLE[sync_data_threads_.size()];
    for (size_t i = 0; i < sync_data_threads_.size(); ++i)
    {
        handle[i] = sync_data_threads_[i]->thread();
    }
    WaitForMultipleObjects((int)sync_data_threads_.size(), handle, TRUE, INFINITE);

    delete[]handle;
    sync_data_threads_.clear();

    if (task_index_ < tasks_.size())
    {
        syncData(sync_id, vault, enclave_provider);
    }
}

bool SyncData::getTask(Task & task)
{
	if (task_index_ >= tasks_.size())
	{
		return false;
	}

	task = tasks_[task_index_];

	++task_index_;

	return true;
}

void SyncData::onNotifySync(const::enclave::SyncNotifyRequest *request, ::enclave::SyncTablesDatas *response)
{
    HanaData hana(config_->database_);

    LOG(INFO) << "Load sync cache data";
    sync_data_cache_.load();

	auto tables_datas = sync_data_cache_.tables_datas();

    for (int i = 0; i < request->tables_size(); i++)
    {
        const ::enclave::Table& table = request->tables(i);

        std::vector<Field> columns;
        for (int j = 0; j < table.columns_size(); j++)
        {
            Field field;
            Util::GColumnToIColumn(table.columns(j), field, true, true);
            columns.push_back(field);
        }

        tables_columns_[table.name()] = columns;

        if (hana.isTableExists(table.name()) == false)
        {
            LOG(INFO) << "Create table" << table.name();    
            hana.createTable_sync(table.name(), columns);
        }

		if (tables_datas.find(table.name()) == tables_datas.end())
		{
			continue;
		}

        auto table_datas = tables_datas[table.name()];

        auto sync_datas = response->add_tablesyncdatas();
        sync_datas->set_table(table.name());
        sync_datas->set_end(table_datas.end_);

        for (auto iter = table_datas.datas_.begin(); iter != table_datas.datas_.end(); iter++)
        {
            if (iter->is_successed_)
            {
                continue;
            }

            auto sync_data = sync_datas->add_syncdatas();
            sync_data->set_start(iter->start_);
            sync_data->set_end(iter->end_);

            LOG(INFO) << table.name() << ", Cache data start = " << iter->start_ << ", end = " << iter->end_;
        }

        LOG(INFO) << table.name() << ", Cache data count = " << sync_datas->syncdatas_size();
    }
}

void SyncData::onSyncInfo(const::enclave::SyncTablesDatas * request)
{
    LOG(INFO) << "Add sync task";
    sync_data_cache_.addData(*request);

	LOG(INFO) << "Create sync write data thread";

	for (int i = 0; i < config_->thread_number_; i++)
	{
		auto sync_write_data_thread = std::shared_ptr<SyncWriteDataThread>(new SyncWriteDataThread(this, config_, enclave_provider_));

		if (sync_write_data_thread->run())
		{
			sync_write_data_threads_.push_back(sync_write_data_thread);
		}
	}

	LOG(INFO) << "Sync write data number of threads is " << sync_write_data_threads_.size();
}

void SyncData::addSyncDatas(std::shared_ptr<enclave::SyncTableRows> sync_table_rows)
{
	CriticalSectionUtil cs_util(sync_datas_cs_);

	sync_datas_.push_back(sync_table_rows);

	event_.SetEvent();
}

bool SyncData::popupSyncDatas(std::shared_ptr <enclave::SyncTableRows> &sync_table_rows)
{
	CriticalSectionUtil cs_util(sync_datas_cs_);

	if (sync_datas_.size() == 0)
	{
		event_.ResetEvent();
		return false;
	}

	sync_table_rows = sync_datas_.front();

	sync_datas_.pop_front();

	return true;
}

int SyncData::popupSyncDatasSize()
{
	CriticalSectionUtil cs_util(sync_datas_cs_);

	return (int)sync_datas_.size();
}

void SyncData::setSyncFinished(bool finished) 
{
	is_sync_finished_ = finished;

	event_.SetEvent();

	HANDLE *handle = new HANDLE[sync_write_data_threads_.size()];
	for (size_t i = 0; i < sync_write_data_threads_.size(); ++i)
	{
		handle[i] = sync_write_data_threads_[i]->thread();
	}
	WaitForMultipleObjects((int)sync_write_data_threads_.size(), handle, TRUE, INFINITE);

	delete[]handle;
	sync_write_data_threads_.clear();
}
